#include "sd_ttree.h"
#include "sd_memory.h"
#include "sd_log.h"
#include "sd_thread.h"
#include "sd_transaction.h"
#include "sd_oplog.h"
#include "sd_hashtable.h"


typedef enum KeyLocation
{
	KL_Unknown,
	KL_InternalNode,
	KL_SubtreeLeft,
	KL_SubtreeRight,
}KeyLocation;

/* Definitions of the bits in lock code */
#define SD_TTREE_XLOCK_PATH  0x2
#define SD_TTREE_XLOCK_END   0x1
#define SD_TTREE_SLOCK_END   0x0



static sd_TTreeNode* CreateNode(sd_TTree* tree);
static void          FreeNode(sd_TTreeNode* node);
static sd_Ret        InsertToNode(sd_TTreeNode* node, const sd_Record* record, int pos, KeyLocation loc, bool* bNewLeaf);
static sd_Ret        DeleteFromNode(sd_TTreeNode* node, int recordIndex, int count, bool* isNodeFreed, sd_uid* freeNodeID);
static sd_TTreeNode* TraverseNode(sd_TTreeNode* node, const sd_Key* key, KeyLocation* loc, int lockcode);
static sd_TTreeNode* TraverseRecord(sd_TTreeNode* node, const sd_Record* record, int* pos, int lockcode);
static sd_TTreeNode* TraverseLeftMostNode(sd_TTreeNode* node, int lockcode);
static sd_TTreeNode* TraverseRightMostNode(sd_TTreeNode* node, int lockcode);
static sd_TTreeNode* TraverseNextKeyNode(sd_TTreeNode* node, const sd_Key* key);
static sd_Ret        BacktraceNode(sd_TTreeNode* fromNode, sd_TTreeNode* endNode, bool bUpdateTree);
static void          UpdateNodeDepth(sd_TTreeNode* node); 
static sd_Ret        RebalanceNode(sd_TTreeNode** node);
static int           FindKeyFirstPosInNode(sd_TTreeNode* node, int startpos, const sd_Key* key, bool *bExisted);
static sd_TTreeNode* FindRecordWithSameKey(sd_TTreeNode* node, int* pos, const sd_Record* record, int lockcode);
static sd_TTreeNode* MoveToRecordEndPosOfSameKey(sd_TTreeNode* node, int* pos, const sd_Key* key, int lockcode);
static void          PrintNode(sd_TTreeNode* node, FILE* fp, int indent);
static sd_Ret        CheckNode(sd_TTreeNode* node);
static sd_Ret        LockShareNode(sd_TTreeNode* node);
static sd_Ret        LockExclusiveNode(sd_TTreeNode* node);
static sd_Ret        UnlockNode(sd_TTreeNode* node);
static int			 GetDepthDifference(sd_TTreeNode*  node);
static void			 LeftRotation(sd_TTreeNode**  rootNode ,sd_TTreeNode*  pivotNode,  bool pivotNodeLock );
static void			 RightRotation(sd_TTreeNode**  rootNode ,sd_TTreeNode*  pivotNode, bool pivotNodeLock);
static void          CheckLockTable(sd_TTree* tree);


sd_Ret sd_TTreeCreate(sd_TTree** tree, size_t maxRecOfNode, sd_KeyCompare key_cmp)
{
	*tree = (sd_TTree*)sd_malloc(sizeof(sd_TTree));
	if (*tree == NULL)
	{
		sd_log_err("Cannot allocate enough memory !\n");
		SD_ASSERT(false);
		return SD_RET_ALLOCATE_MEMEORY;
	}
	(*tree)->maxNumRecordsOfNode = maxRecOfNode;
	(*tree)->key_cmp = key_cmp;
	(*tree)->rootNode = NULL;
	(*tree)->depth = 0;
#ifdef _DEBUG_TTREE_LOCK
	sd_SpinLock_Init(&(*tree)->tidLock);
	sd_HashtableCreate(&(*tree)->nodeLockTable,
		sd_HashFunc_uint32,
		sd_HashKeyCompare_uint32,
		32,
		8,
		sizeof(uint32),
		sizeof(sd_Hashtable*),
		NULL);
#endif
	SD_UID_ZERO((*tree)->nodeIdCounter);
	sd_SpinLock_Init(&(*tree)->lock);
	return SD_RET_OK;
}

sd_Ret sd_TTreeFree(sd_TTree* tree)
{
	sd_SpinLock_Destroy(&tree->lock);
#ifdef _DEBUG_TTREE_LOCK
	sd_SpinLock_Destroy(&tree->tidLock);
	sd_HashtableFree(tree->nodeLockTable);
#endif
	FreeNode(tree->rootNode);
	sd_free(tree);
	return SD_RET_OK;
}

sd_Ret sd_TTreeInsert(sd_TTree* tree, const sd_Record* record)
{
	sd_Ret ret;
	sd_SpinLock_Lock(&tree->lock);
	if (tree->rootNode == NULL)
	{
		sd_TTreeNode* rootNode = CreateNode(tree);
		tree->rootNode = rootNode;
		memcpy(&tree->rootNode->records[0], record, sizeof(sd_Record));
		tree->rootNode->numRecords++;
		tree->rootNode->depth = 1;
		tree->depth = 1;
		sd_SpinLock_Unlock(&tree->lock);  /* Unlock this structure of this tree */
	}
	else
	{	
		KeyLocation loc, oldloc;
		int pos = 0;
		bool bKeyExisted;
		bool bCreateNewLeaf;
		sd_TTreeNode* node;
		sd_uid oldnodeid;
		int lockcode = SD_TTREE_XLOCK_END;
		sd_SpinLock_Unlock(&tree->lock);  /* Unlock this structure of this tree */

#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif

_BEGIN_TRAVERSE_NODE:
		node = TraverseNode(tree->rootNode,  &record->key, &loc, lockcode);
		SD_ASSERT(node != NULL);
		oldloc = loc;
		oldnodeid = node->id;

		if (loc == KL_InternalNode)
		{
			/* Find the key's position in this node */
			pos = FindKeyFirstPosInNode(node, pos, &record->key, &bKeyExisted);
			if (bKeyExisted)
			{
				sd_TTreeNode* nodeWithDuplicateRecord;
				int temppos = pos;
				/* Check this node contains the duplicate record */
				nodeWithDuplicateRecord = FindRecordWithSameKey(node, &temppos, record, SD_TTREE_SLOCK_END);
				if (nodeWithDuplicateRecord !=  NULL)
				{
					/* Found the duplicate record */
					UnlockNode(nodeWithDuplicateRecord);
					if (lockcode & SD_TTREE_XLOCK_PATH)
					{
						BacktraceNode(node, tree->rootNode, false);
					}
					else
					{
						UnlockNode(node);
					}
#ifdef _DEBUG_TTREE_LOCK
					CheckLockTable(tree);
#endif
					return SD_SET_RECORD_EXISTS;
				}
				
				/* Find the records' end position of the same key */
				node = MoveToRecordEndPosOfSameKey(node, &pos, &record->key, lockcode);
				if (pos == tree->maxNumRecordsOfNode)
				{
					loc = KL_SubtreeRight;
				}
			}
		}
        
		/* If this node is not safe to insert */
		if ((lockcode & SD_TTREE_XLOCK_PATH) == 0)
		{
			if (node->numRecords == tree->maxNumRecordsOfNode||	loc != KL_InternalNode)
			{
				/* Unlock this node */
				UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
				CheckLockTable(tree);
#endif
				/* Re-traverse to this node, x-locked all its ancestors */
				lockcode = SD_TTREE_XLOCK_END|SD_TTREE_XLOCK_PATH;
				goto _BEGIN_TRAVERSE_NODE;
			}
		}

		/* Insert this record to the node */
		ret = InsertToNode(node, record, pos, loc, &bCreateNewLeaf);
		SD_ON_FAILED_RETURN(ret);

		/* Unlock all nodes' locks */
		if (lockcode & SD_TTREE_XLOCK_PATH)
		{
			BacktraceNode(node, tree->rootNode, bCreateNewLeaf);
		}
		else
		{
			UnlockNode(node);
		}
#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif
	}

	return SD_RET_OK;
}

sd_Ret sd_TTreeGetFirst(sd_TTree* tree, sd_Record* record)
{
	sd_TTreeNode* node = NULL;

	if (tree->rootNode == NULL)
		return SD_RET_KEY_NOTFOUND;

	node = TraverseLeftMostNode(tree->rootNode, SD_TTREE_SLOCK_END);
	if (node->numRecords == 0)
	{
		UnlockNode(node);
		return SD_RET_KEY_NOTFOUND;
	}
	else
	{
		memcpy(record, &node->records[0], sizeof(sd_Record));
		UnlockNode(node);
		return SD_RET_OK;
	}
}

sd_Ret sd_TTreeSearch(sd_TTree* tree, const sd_Key* key, sd_Record* record)
{
	KeyLocation loc;
	sd_TTreeNode* node = NULL;
	bool bKeyExisted;

	if (tree->rootNode == NULL)
		return SD_RET_KEY_NOTFOUND;

	node = TraverseNode(tree->rootNode, key, &loc, SD_TTREE_SLOCK_END);
	if (loc == KL_InternalNode)
	{
		int pos;
		pos = FindKeyFirstPosInNode(node, 0, key, &bKeyExisted);
		if (!bKeyExisted)
		{
			/* Not found */
			UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
			CheckLockTable(tree);
#endif
			return SD_RET_KEY_NOTFOUND;
		}
		else
		{
			/* Found this key */
			memcpy(record, &node->records[pos], sizeof(sd_Record));
			/* Not found */
			UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
			CheckLockTable(tree);
#endif
			return SD_RET_OK;
		}
	}
	else
	{
		/* Not found */
		UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif
		return SD_RET_KEY_NOTFOUND;
	}
}

sd_Ret sd_TTreeSearchNext(sd_TTree* tree, const sd_Record* preRecord, sd_Record* record)
{
	sd_Ret ret;
	sd_TTreeNode* parentNode=NULL;
	if (tree->rootNode == NULL)
	{
		return SD_RET_DB_END;
	}
	else
	{
		int pos = 0;
		bool bNeedFindNextKey = false;
		sd_Key currentKey;
		sd_TTreeNode* node;
		sd_TTreeNode* leftMostNode;
		sd_uid preid;

#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif
		
		if (preRecord->payload[0] != 0) /* in the last time, the call 'get' return is NOT KEY_NOTFOUND */
		{
			/* Find the node contains the record */
			node = TraverseRecord(tree->rootNode,  preRecord, &pos, SD_TTREE_SLOCK_END);
			if (node == NULL)
			{
				currentKey = preRecord->key;
				bNeedFindNextKey = true; 
			}
			else
			{
				preid = node->id;
			}

			if (bNeedFindNextKey)
			{
				node = TraverseNextKeyNode(tree->rootNode, &currentKey);
				if (node == NULL)
				{
#ifdef _DEBUG_TTREE_LOCK
					CheckLockTable(tree);
#endif
					return SD_RET_DB_END;
				}
				memcpy(record, &node->records[0], sizeof(sd_Record));
			}
			else
			{
				pos++;
				if (pos >= (int)node->numRecords)
				{
					
					if (node->rightChild == NULL)
					{
						currentKey = node->records[pos-1].key;
						UnlockNode(node);
						node = TraverseNextKeyNode(tree->rootNode, &currentKey);
						if (node == NULL)
						{
#ifdef _DEBUG_TTREE_LOCK
							CheckLockTable(tree);
#endif
							return SD_RET_DB_END;
						}
						memcpy(record, &node->records[0], sizeof(sd_Record));
					}
					else
					{
						leftMostNode = TraverseLeftMostNode(node->rightChild, SD_TTREE_SLOCK_END);
						SD_ASSERT(leftMostNode != NULL);
						UnlockNode(node);
						node = leftMostNode;
						memcpy(record, &node->records[0], sizeof(sd_Record));
					}
				}
				else
				{
					memcpy(record, &node->records[pos], sizeof(sd_Record));
				}
			}
			/* Release the lock */
			UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
			CheckLockTable(tree);
#endif
		}
		else /* in the last time, the call 'get' return a KEY_NOTFOUND */
		{
			bool bExisted;
			
			/* Find the node contains the key */
			ret = sd_TTreeSearch(tree, &preRecord->key, record);
			if (ret == SD_RET_KEY_NOTFOUND)
			{
				/* Find the node contains a larger key */
				node = TraverseNextKeyNode(tree->rootNode, &preRecord->key);
				if (node == NULL)
				{
					return SD_RET_DB_END;
				}
				pos = FindKeyFirstPosInNode(node, 0, &preRecord->key, &bExisted);
				while(pos < (int)node->numRecords-1 && tree->key_cmp(&node->records[pos].key, &preRecord->key)<=0)
					pos++;
				memcpy(record, &node->records[pos], sizeof(sd_Record));

				/* Release the lock */
				UnlockNode(node);
#ifdef _DEBUG_TTREE_LOCK
				CheckLockTable(tree);
#endif
			}
		}

	}

	return SD_RET_OK;
}

sd_Ret sd_TTreeSearchRID(sd_TTree* tree, sd_Record* record, sd_uid* rid)
{
	bool bFound = false;
	sd_TTreeNode* parentNode=NULL;
	if (tree->rootNode == NULL)
	{
		return SD_RET_KEY_NOTFOUND;
	}
	else
	{
		int pos = 0;

		/* Find the node contains the record */
		sd_TTreeNode* node = TraverseRecord(tree->rootNode, record, &pos, SD_TTREE_SLOCK_END);
		if (node == NULL)
		{
#ifdef _DEBUG_TTREE_LOCK
			CheckLockTable(tree);
#endif
			return SD_RET_KEY_NOTFOUND;
		}

		/* Copy the rid and record to return */
		*rid = node->records[pos].rid;
		memcpy(record, &node->records[pos], sizeof(sd_Record));

		/* Release the lock */
		UnlockNode(node);

#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif

		return SD_RET_OK;
	}		
}

sd_Ret sd_TTreeDelete(sd_TTree* tree,  const sd_Record* record)
{
	sd_Ret ret;
	int numDeleted = 0;
	bool isNodeDeleted;
	sd_TTreeNode* parentNode=NULL;
	if (tree->rootNode == NULL)
	{
		return SD_RET_KEY_NOTFOUND;
	}
	else
	{	
		bool bLeaf;
		sd_uid nodeId;
		sd_uid freeNodeId;
		int pos = 0;
		bool bXLockedAllAncestors = false;

#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif

		/* Find the node contains the record */
		sd_TTreeNode* node = TraverseRecord(tree->rootNode,  record, &pos, SD_TTREE_XLOCK_END);
		if (node == NULL)
			return SD_RET_KEY_NOTFOUND;

		bLeaf = (node->leftChild == NULL) && (node->rightChild == NULL);

		/* If this node is not safe */
		if ((bLeaf && node->numRecords <= 1) || (node->numRecords-1) < tree->maxNumRecordsOfNode/2) 
		{
			UnlockNode(node);
			node = TraverseRecord(tree->rootNode, record, &pos, SD_TTREE_XLOCK_PATH | SD_TTREE_XLOCK_END);
			if (node == NULL)
				return SD_RET_KEY_NOTFOUND;
			bXLockedAllAncestors = true;
		}

		/* Delete the record from the record */
		nodeId = node->id;
		numDeleted++;
		parentNode=node->parent;
		ret = DeleteFromNode(node, pos, 1, &isNodeDeleted, &freeNodeId);
		if(isNodeDeleted && nodeId == freeNodeId)
		{
			node=parentNode;
		}
		SD_ON_FAILED_RETURN(ret);

		/* Release all the locks */
		if (bXLockedAllAncestors)
			BacktraceNode(node, tree->rootNode,isNodeDeleted);
		else
			UnlockNode(node);

#ifdef _DEBUG_TTREE_LOCK
		CheckLockTable(tree);
#endif
	}

	if (numDeleted == 0)
		return SD_RET_KEY_NOTFOUND;
	else
		return SD_RET_OK;
}

sd_Ret sd_TTreePrint(sd_TTree* tree, FILE* fp)
{
	PrintNode(tree->rootNode, fp, 0);
	return SD_RET_OK;
}


sd_Ret sd_TTreeCheck(sd_TTree* tree)
{
	sd_TTreeNode* rootNode = tree->rootNode;
	return CheckNode(rootNode);
}


static sd_TTreeNode* CreateNode(sd_TTree* tree)
{
	sd_TTreeNode* newNode;

	newNode = (sd_TTreeNode*)sd_malloc(sizeof(sd_TTreeNode)+(tree->maxNumRecordsOfNode-1)*sizeof(sd_Record));
	if (newNode == NULL)
	{
		sd_log_err("Cannot allocate enough memory !\n");
		SD_ASSERT(false);
		return NULL;
	}
	newNode->leftChild = NULL;
	newNode->rightChild = NULL;
	newNode->parent = NULL;
	newNode->numRecords = 0;
	newNode->tree = tree;
	newNode->depth = 1;
	if (tree->rootNode != NULL)
	{
		sd_SpinLock_Lock(&tree->lock);
		newNode->id = tree->nodeIdCounter;
		SD_UID_INC(tree->nodeIdCounter);
		sd_SpinLock_Unlock(&tree->lock);
	}
	else
	{
		newNode->id = tree->nodeIdCounter;
		SD_UID_INC(tree->nodeIdCounter);
	}

	sd_RWLock_Init(&newNode->lock);
	return newNode;
}

static void FreeNode(sd_TTreeNode* node)
{
	if (node != NULL)
	{
		FreeNode(node->leftChild);
		FreeNode(node->rightChild);
		/* Delete this node */
		UnlockNode(node);
		sd_RWLock_Destroy(&node->lock);
		sd_free(node);
	}
}

static sd_TTreeNode* TraverseNode(sd_TTreeNode* node, const sd_Key* key, KeyLocation* loc, int lockcode)
{
	int cmp_min_ret,cmp_max_ret;
	sd_TTree* tree = node->tree;

	if (lockcode != 0)
		LockExclusiveNode(node);
	else
		LockShareNode(node);

	while(true) /* node is not the end node */
	{
		if (node->numRecords == 0)
		{
			*loc= KL_InternalNode;
			break;
		}
		cmp_min_ret = tree->key_cmp(key, &node->records[0].key);
		cmp_max_ret = tree->key_cmp(key, &node->records[node->numRecords-1].key);
		if (cmp_min_ret >=0 && cmp_max_ret <= 0)
		{
			*loc = KL_InternalNode;
			break;
		}
		if (cmp_min_ret < 0)
		{
			if (node->leftChild == NULL)
			{
				*loc = KL_SubtreeLeft;
				break;
			}
			else
			{
				if (lockcode & SD_TTREE_XLOCK_PATH)
				{
					LockExclusiveNode(node->leftChild);
				}
				else if (lockcode & SD_TTREE_XLOCK_END)
				{
					LockExclusiveNode(node->leftChild);
					UnlockNode(node);
				}
				else
				{
					LockShareNode(node->leftChild);
					UnlockNode(node);
				}
				node = node->leftChild;
			}
		}
		else if (cmp_max_ret > 0)
		{
			if (node->rightChild == NULL)
			{
				*loc = KL_SubtreeRight;
				break;
			}
			else
			{
				if (lockcode & SD_TTREE_XLOCK_PATH)
				{
					LockExclusiveNode(node->rightChild);
				}
				else if (lockcode & SD_TTREE_XLOCK_END)
				{
					LockExclusiveNode(node->rightChild);
					UnlockNode(node);
				}
				else
				{
					LockShareNode(node->rightChild);
					UnlockNode(node);
				}
				node = node->rightChild;
			}
		}
	}

	return node;
}

static sd_TTreeNode* TraverseRecord(sd_TTreeNode* node, const sd_Record* record, int* pos, int lockcode)
{
	sd_TTree* tree = node->tree;
	sd_TTreeNode* originalNode = node;
	bool bContainPayload = record->payload[0] != 0;
	KeyLocation loc;
	bool bKeyExisted;

	*pos = 0;

	/* Find the node contains the key of the record */
	node = TraverseNode(node, &record->key, &loc, lockcode);
	SD_ASSERT(node != NULL);

	if (loc != KL_InternalNode)
	{
		if (lockcode & SD_TTREE_XLOCK_PATH)
			BacktraceNode(node, originalNode, false);
		else
			UnlockNode(node);
		return NULL;
	}

	/* Find the key's position in this node */
	*pos = FindKeyFirstPosInNode(node, *pos, &record->key, &bKeyExisted);
	if (!bKeyExisted)
	{
		if (lockcode & SD_TTREE_XLOCK_PATH)
			BacktraceNode(node, originalNode, false);
		else
			UnlockNode(node);
		return NULL;
	}

	if (bContainPayload) /* In this case, we need to compare the payload between records */
	{
		sd_TTreeNode* finalNode = FindRecordWithSameKey(node, pos, record, lockcode);
		if (finalNode == NULL)
		{
			if (lockcode & SD_TTREE_XLOCK_PATH)
				BacktraceNode(node, originalNode, false);
			else
				UnlockNode(node);
			return NULL;
		}
		else
		{
			if (finalNode != node && !(lockcode & SD_TTREE_XLOCK_PATH))
				UnlockNode(node);
			return finalNode;
		}
	}
	else /* In this case, the user didn't specified the payload of the deleting records */ 
	{
		return node;
	}
}

static sd_TTreeNode* TraverseNextKeyNode(sd_TTreeNode* node, const sd_Key* key)
{
	int cmp_min_ret,cmp_max_ret;
	sd_TTree* tree = node->tree;
	sd_TTreeNode* largerNode = NULL;

	LockShareNode(node);

	while(true) /* node is not the end node */
	{
		cmp_min_ret = tree->key_cmp(key, &node->records[0].key);
		cmp_max_ret = tree->key_cmp(key, &node->records[node->numRecords-1].key);
		if (cmp_min_ret >=0 && cmp_max_ret < 0)
		{
			if (largerNode != NULL)
				UnlockNode(largerNode);

			largerNode = node;
			break;
		}
		if (cmp_min_ret < 0)
		{
			if (largerNode != NULL)
				UnlockNode(largerNode);
			largerNode = node;
			if (node->leftChild == NULL)
			{
				break;
			}
			else
			{
				LockShareNode(node->leftChild);
				node = node->leftChild;
			}
		}
		else if (cmp_max_ret >= 0)
		{
			if (node->rightChild == NULL)
			{
				/* Not find any other key larger than the given key */
				UnlockNode(node);
				break;
			}
			else
			{
				LockShareNode(node->rightChild);
				UnlockNode(node);
				node = node->rightChild;
			}
		}
	}
	return largerNode;
}

static sd_Ret BacktraceNode(sd_TTreeNode* fromNode, sd_TTreeNode* endNode, bool bUpdateTree)
{
	sd_TTreeNode* node = fromNode;
	sd_TTreeNode* parent = NULL;
    sd_TTreeNode** nodePointer=NULL;
	while(node != NULL)
	{
	
		parent = node->parent;
        /*if(parent==NULL)
		{
			sd_SpinLock_Lock(&node->tree->lock);
		
		}*/
		if (bUpdateTree)
		{
			/* Update the node's depth */
			UpdateNodeDepth(node);
			/* Rebalance this node */
			/*nodePointer=&node;
			if(node->parent!=NULL)
				RebalanceNode(nodePointer);
			node=*nodePointer;*/
			
		}

		/* After finshied all the jobs with the node, unlock this node */
	
		
		UnlockNode(node);
		/*if(node->parent==NULL)
		{
			node->tree->rootNode = node;
			sd_SpinLock_Unlock(&node->tree->lock);
		}*/
		
		
		if (node == endNode)
			break;
		node = parent;
	}
	return SD_RET_OK;
}

static sd_Ret InsertToNode(sd_TTreeNode* node, const sd_Record* record, int pos, KeyLocation loc, bool* bNewLeaf)
{
	sd_Ret ret;
	sd_TTree* tree = node->tree;
	int movecount;
	sd_TTreeNode* newNode = NULL;

	if (node->numRecords == tree->maxNumRecordsOfNode) /* If this node's record array is full */
	{
		switch (loc)
		{
		case KL_SubtreeLeft:
			{
				/* The left child node must be NULL */
				SD_ASSERT(node->leftChild == NULL);

				/* Create new left child node */
				newNode = CreateNode(tree);
				newNode->parent = node;
				newNode->records[0] = *record;
				newNode->numRecords = 1;
				newNode->depth = 1;
				node->leftChild = newNode;
				*bNewLeaf =true;
				break;
			}
		case KL_SubtreeRight:
			{
				/* The right child node must be NULL */
				SD_ASSERT(node->rightChild == NULL);

				/* Create new right child node */
				newNode = CreateNode(tree);
				newNode->parent = node;
				newNode->records[0] = *record;
				newNode->numRecords = 1;
				newNode->depth = 1;
				node->rightChild = newNode;
				*bNewLeaf =true;
				break;
			}
		case KL_InternalNode:
			{
				/* Move the largest record to another node to make a space for current inserted record */
				if (node->rightChild != NULL)
				{
					/* Move the maximum record to the left most or right subtree */
					sd_TTreeNode* leftMostNode = TraverseLeftMostNode(node->rightChild, SD_TTREE_XLOCK_END);
					if (leftMostNode->numRecords < tree->maxNumRecordsOfNode)
					{
						/* Insert the largest record to the left most node of the right subtree */
						ret = InsertToNode(leftMostNode, &node->records[node->numRecords-1], 0, KL_InternalNode, bNewLeaf);
						SD_ON_FAILED_RETURN(ret);

						UnlockNode(leftMostNode);
					}
					else
					{
						UnlockNode(leftMostNode);

						/* Traverse the left most node and x-lock its ancestors to the current node*/ 
						leftMostNode = TraverseLeftMostNode(node->rightChild, SD_TTREE_XLOCK_PATH|SD_TTREE_XLOCK_END);

						newNode = CreateNode(tree);
						newNode->parent = leftMostNode;
						newNode->records[0] = node->records[node->numRecords-1];
						newNode->numRecords = 1;
						newNode->depth = 1;
						leftMostNode->leftChild=newNode;
						*bNewLeaf = true;
						ret = BacktraceNode(leftMostNode, node->rightChild, true);
						SD_ON_FAILED_RETURN(ret);
					}
					node->numRecords--;
				}
				else /* This node's right child node is NULL */
				{
					/* Create new right child node */
					newNode = CreateNode(tree);
					newNode->parent = node;
					newNode->records[0] = node->records[node->numRecords-1];
					newNode->numRecords = 1;
					newNode->depth = 1;
					node->numRecords--;
					node->rightChild = newNode;
					*bNewLeaf =true;
				}

				/* Insert the record to the current node */
				movecount = node->numRecords-pos;
				if (movecount > 0)
				{
					memmove(&node->records[pos+1], &node->records[pos], (size_t)(movecount*sizeof(sd_Record)));
				}
				memcpy(&node->records[pos], record, sizeof(sd_Record));
				node->numRecords++;
				break;
			}
		default:
			SD_ASSERT(false);
			break;
		}
	}
	else /* If this node's record array is not full */
	{
		switch (loc)
		{
		case KL_SubtreeLeft:
			pos = 0;
			break;
		case KL_SubtreeRight:
			pos = node->numRecords;
			break;
		}			
		movecount = node->numRecords-pos;
		if (movecount > 0)
		{
			memmove(&node->records[pos+1], &node->records[pos], (size_t)(movecount*sizeof(sd_Record)));
		}
		memcpy(&node->records[pos], record, sizeof(sd_Record));
		node->numRecords++;
		*bNewLeaf = false;
	}
	return SD_RET_OK;
}


static sd_Ret DeleteFromNode(sd_TTreeNode* node, int recordIndex,int count, bool* isNodeFreed, sd_uid* freedNodeId)
{
	sd_Ret ret;
	sd_TTree* tree = node->tree;
	int movecount;

	SD_ASSERT(node->numRecords >= (size_t)(recordIndex+count));

	movecount = (int)node->numRecords - recordIndex - count;
	if (movecount > 0)
	{
		memmove(&node->records[recordIndex],&node->records[recordIndex+count], movecount*sizeof(sd_Record));
	}
	node->numRecords -= count;

	/* Check the underflow of this node */
	*isNodeFreed = false;
	if (node->leftChild == NULL && node->rightChild == NULL) /* leaf node */
	{
		if (node->numRecords == 0)
		{
			sd_TTreeNode* parentNode = node->parent;
			if (parentNode != NULL)
			{
				if (parentNode->leftChild == node)
					parentNode->leftChild = NULL;
				else
					parentNode->rightChild = NULL;
			}
			else
			{
				return SD_RET_OK;
			}
			
			*freedNodeId = node->id;
			FreeNode(node);
			*isNodeFreed = true;
		}
	}
	else /* Not a leaf node */
	{
		/* Move records from left or right most node to the current node */
		int movepos;
		size_t maxNumRecords = node->tree->maxNumRecordsOfNode;
		if (node->numRecords < maxNumRecords/2) /* The record count is underflow */
		{
			if (node->rightChild != NULL)
			{
				sd_TTreeNode* leftMostNode = NULL;
				sd_TTreeNode* leftMNParent = NULL;
				sd_uid leftMostNodeID;
				/* Move the records from the left most right-subtree to current node */
				leftMostNode = TraverseLeftMostNode(node->rightChild, SD_TTREE_XLOCK_END);
				movecount = MIN(leftMostNode->numRecords, maxNumRecords - node->numRecords);
				if (leftMostNode->numRecords - movecount < maxNumRecords/2)  /* The left most node is not safe */
				{
					/* Re-traverse the left most node with x-locking all ancestors */
					UnlockNode(leftMostNode);
					leftMostNode = TraverseLeftMostNode(node->rightChild, SD_TTREE_XLOCK_PATH|SD_TTREE_XLOCK_END);
					movecount = MIN(leftMostNode->numRecords, maxNumRecords - node->numRecords);

					/* Move the records */
					memcpy(&node->records[node->numRecords], &leftMostNode->records[0], movecount*sizeof(sd_Record));
					node->numRecords += movecount;
					
					leftMNParent = leftMostNode->parent;
					leftMostNodeID = leftMostNode->id;
					ret = DeleteFromNode(leftMostNode, 0, movecount, isNodeFreed, freedNodeId);
					SD_ON_FAILED_RETURN(ret);
					/* Backtrace from the left most node to the current node, and unlock these nodes */
					if(*isNodeFreed && *freedNodeId == leftMostNodeID)
					{
						if (leftMNParent != node)
						{
							ret = BacktraceNode(leftMNParent, node->rightChild, true);
							SD_ON_FAILED_RETURN(ret);
						}
					}
					else
					{
						ret = BacktraceNode(leftMostNode, node->rightChild, *isNodeFreed);
						SD_ON_FAILED_RETURN(ret);
					}
				}
				else  /* The left most node is safe */
				{
					/* Move the records */
					memcpy(&node->records[node->numRecords], &leftMostNode->records[0], movecount*sizeof(sd_Record));
					node->numRecords += movecount;

					ret = DeleteFromNode(leftMostNode, 0, movecount, isNodeFreed, freedNodeId);
					SD_ON_FAILED_RETURN(ret);
					SD_ASSERT(*isNodeFreed == false);
					/* Unlock the left most node */
					UnlockNode(leftMostNode);
				}
			}
			else if(node->leftChild != NULL)
			{
				sd_TTreeNode* rightMostNode;
				sd_TTreeNode* rightMNParent = NULL;
				sd_uid rightMostNodeID;
				/* Move the records from the right most left-subtree to current node */
				rightMostNode = TraverseRightMostNode(node->leftChild, SD_TTREE_XLOCK_END);
				movecount = MIN(rightMostNode->numRecords, maxNumRecords - node->numRecords);
				if (rightMostNode->numRecords - movecount < maxNumRecords/2) /* The right most node is not safe */
				{
					/* Re-traverse the right most node with x-locking all ancestors */
					UnlockNode(rightMostNode);
					rightMostNode = TraverseRightMostNode(node->leftChild,  SD_TTREE_XLOCK_PATH|SD_TTREE_XLOCK_END);
					movecount = MIN(rightMostNode->numRecords, maxNumRecords - node->numRecords);

					/* Move the records */
					movepos = (int)(rightMostNode->numRecords - movecount);
					memmove(&node->records[movecount], &node->records[0], movecount*sizeof(sd_Record));
					memcpy(&node->records[0], &rightMostNode->records[movepos], movecount*sizeof(sd_Record));
					node->numRecords += movecount;

					rightMNParent=rightMostNode->parent;
					rightMostNodeID = rightMostNode->id;
					ret = DeleteFromNode(rightMostNode, movepos, movecount, isNodeFreed, freedNodeId);

					/* Backtrace from the right most node to the current node, and unlock these nodes */
					if(*isNodeFreed && *freedNodeId == rightMostNodeID)
					{
						if (rightMNParent != node)
						{
							ret = BacktraceNode(rightMNParent, node->leftChild, true);
							SD_ON_FAILED_RETURN(ret);
						}
					}
					else
					{
						ret = BacktraceNode(rightMostNode, node->leftChild, *isNodeFreed);
						SD_ON_FAILED_RETURN(ret);
					}
				}
				else /* The right most node is safe */
				{
					/* Move the records */
					movepos = (int)(rightMostNode->numRecords - movecount);
					memmove(&node->records[movecount], &node->records[0], movecount*sizeof(sd_Record));
					memcpy(&node->records[0], &rightMostNode->records[movepos], movecount*sizeof(sd_Record));
					node->numRecords += movecount;

					ret = DeleteFromNode(rightMostNode, movepos, movecount, isNodeFreed, freedNodeId);
					SD_ON_FAILED_RETURN(ret);
					SD_ASSERT(*isNodeFreed == false);
					/* Unlock the left most node */
					UnlockNode(rightMostNode);
				}
			}

		}
	}
	return SD_RET_OK;
}

static int FindKeyFirstPosInNode(sd_TTreeNode* node, int startpos, const sd_Key* key, bool *bExisted)
{
	sd_TTree* tree = node->tree;
	int lb,ub,mid,pos;
	int cmp_ret;
	sd_Key* midKey;
	lb = startpos;
	ub = node->numRecords-1;
	*bExisted = false;

	if (node->numRecords == 0)
		return 0;

	cmp_ret = tree->key_cmp(key, &node->records[ub].key);
	if (cmp_ret == 0)
	{
		pos = ub;
		*bExisted = true;
	}
	else
	{
		pos = -1;
		while (lb < ub)
		{
			mid = (lb+ub)/2;
			midKey = &node->records[mid].key;
			cmp_ret = tree->key_cmp(key, midKey);
			if (cmp_ret > 0) /* The key of the given record is larger than the middle key */
			{
				if (lb == mid)
					break;
				else
					lb = mid;
			}
			else if (cmp_ret < 0)
			{
				if (ub == mid)
					break;
				else
					ub = mid;
			}
			else
			{
				pos = mid;
				*bExisted = true;
				break;
			}
		}
		if (pos == -1)
		{
			pos = ub;
			*bExisted = false;
		}
	}

	if (*bExisted)
	{
		/* Let the 'pos' locate the first record with the key */
		while (pos >= 1 && tree->key_cmp(&node->records[pos-1].key, key) == 0)
			pos--;
	}
	if (pos == -1)
	{
		SD_ASSERT(pos);
	}
	return pos;
}

static sd_TTreeNode* FindRecordWithSameKey(sd_TTreeNode* node, int* pos, const sd_Record* record, int lockcode)
{
	sd_TTree* tree = node->tree;
	sd_TTreeNode* originalNode = node;
	while (tree->key_cmp(&node->records[*pos].key, &record->key) == 0)
	{
		if(strcmp(node->records[*pos].payload, record->payload) ==0)
			return node;

		(*pos)++;
		if ((*pos) == node->numRecords)
		{
			if (node->rightChild == NULL)
				break;
			if (lockcode & SD_TTREE_XLOCK_PATH)
				LockExclusiveNode(node->rightChild);
			else if (lockcode & SD_TTREE_XLOCK_END)
			{
				LockExclusiveNode(node->rightChild);
				if (node != originalNode)
					UnlockNode(node);
			}
			else
			{
				LockShareNode(node->rightChild);
				if (node != originalNode)
					UnlockNode(node);
			}
			node = node->rightChild;
			*pos = 0;
		}
	}
	/* If didn't find the given record */
	if (lockcode & SD_TTREE_XLOCK_PATH)
	{
		if (node != originalNode)
		{
			BacktraceNode(node, originalNode->rightChild, false);
		}
	}
	else
	{
		if (node != originalNode)
			UnlockNode(node);
	}
	return NULL;
}

static sd_TTreeNode* MoveToRecordEndPosOfSameKey(sd_TTreeNode* node, int* pos, const sd_Key* key, int lockcode)
{
	sd_TTree* tree = node->tree;
	while (tree->key_cmp(&node->records[*pos].key, key) == 0)
	{
		(*pos)++;
		if (*pos == node->numRecords)
		{
			if (node->rightChild == NULL)
				break;
			if (lockcode & SD_TTREE_XLOCK_PATH)
				LockExclusiveNode(node->rightChild);
			else if (lockcode & SD_TTREE_XLOCK_END)
			{
				LockExclusiveNode(node->rightChild);
				UnlockNode(node);
			}
			else
			{
				LockShareNode(node->rightChild);
				UnlockNode(node);
			}
			node = node->rightChild;
			*pos = 0;
		}
	}
	return node;
}

static sd_TTreeNode* TraverseLeftMostNode(sd_TTreeNode* node, int lockcode)
{

	if (lockcode != 0)
		LockExclusiveNode(node);
	else
		LockShareNode(node);

	while(node->leftChild != NULL)
	{
		if (lockcode & SD_TTREE_XLOCK_PATH)
		{
			LockExclusiveNode(node->leftChild);
		}
		else if (lockcode & SD_TTREE_XLOCK_END)
		{
			LockExclusiveNode(node->leftChild);
			UnlockNode(node);
		}
		else
		{
			LockShareNode(node->leftChild);
			UnlockNode(node);
		}
		node=node->leftChild;
	}
	return node;
}

static sd_TTreeNode* TraverseRightMostNode(sd_TTreeNode* node,  int lockcode)
{
	if (lockcode != 0)
		LockExclusiveNode(node);
	else
		LockShareNode(node);

	while(node->rightChild != NULL)
	{
		if (lockcode & SD_TTREE_XLOCK_PATH)
		{
			LockExclusiveNode(node->rightChild);
		}
		else if (lockcode & SD_TTREE_XLOCK_END)
		{
			LockExclusiveNode(node->rightChild);
			UnlockNode(node);
		}
		else
		{
			LockShareNode(node->rightChild);
			UnlockNode(node);
		}
		node=node->rightChild;
	}
	return node;
}

static void UpdateNodeDepth(sd_TTreeNode* node)
{
	int leftdepth, rightdepth;

	if (node == NULL)
		return;

	//if (node->leftChild != NULL)
	//	LockShareNode(node->leftChild);
	//if (node->rightChild != NULL)
	//	LockShareNode(node->rightChild);

	leftdepth = node->leftChild == NULL ? 0: node->leftChild->depth;
	rightdepth = node->rightChild == NULL ? 0: node->rightChild->depth;

	//if (node->leftChild != NULL)
	//	UnlockNode(node->leftChild);
	//if (node->rightChild != NULL)
	//	UnlockNode(node->rightChild);

	node->depth = MAX(leftdepth, rightdepth)+1;
}

static sd_Ret RebalanceNode(sd_TTreeNode** nodePointer)
{
    sd_TTreeNode**  tempRootNode=NULL;
	sd_TTreeNode* rootNode=* nodePointer;
	sd_TTree* tree = NULL; 
    sd_TTreeNode*  pivotNode=NULL;
	if ( rootNode == NULL)
		return SD_RET_OK;

	tree = rootNode->tree;

	/*
	* Note: There are 4 cases to rotation, please consider the information from :
	* http://en.wikipedia.org/wiki/Tree_rotation
	*/

	/* Left Rotation */
   
    

	if (GetDepthDifference(rootNode)== 2 )
	{
		/* X-lock the node->rightChild */
		LockExclusiveNode(rootNode->rightChild);
		sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),rootNode->rightChild->id);
		if (GetDepthDifference(rootNode->rightChild)==1) /* Right Right Case */
		{
				
			LeftRotation(&rootNode,rootNode->rightChild, false);

		}
		else /* Right Left Case */
		{
		
		    //tempRootNode=&(rootNode->rightChild);
			LockExclusiveNode(rootNode->rightChild->leftChild);

			sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),rootNode->rightChild->leftChild->id);
			RightRotation(&(rootNode->rightChild),rootNode->rightChild->leftChild, false);
			LeftRotation(&rootNode,rootNode->rightChild, false); 
		
		
		}
	 

	}

	/* Right Rotation */
	if (GetDepthDifference(rootNode)== -2 )
	{
		/* X-lock the (*nodePointer)->leftChild */
		LockExclusiveNode(rootNode->leftChild);
		sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),rootNode->leftChild->id);
		if (GetDepthDifference(rootNode->leftChild) ==-1) /* Left Left Case */
		{
		
			
			RightRotation(&rootNode,rootNode->leftChild, false);
		
			
		}
		else  /* Left Right Case */
		{
			
			
			//tempRootNode=&(rootNode->leftChild); 
			LockExclusiveNode(rootNode->leftChild->rightChild);
			
			sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),rootNode->leftChild->rightChild->id);
			LeftRotation(&(rootNode->leftChild), rootNode->leftChild->rightChild, false);
	
			RightRotation(&rootNode,rootNode->leftChild, false);  
		
		}
		
	}

	*nodePointer=rootNode;
	return SD_RET_OK;
}

static int GetDepthDifference(sd_TTreeNode*  node)
{
	int leftDepth; 
	int rightDepth;
    
	leftDepth  = (node->leftChild == NULL) ? 0:node->leftChild->depth;
	rightDepth = (node->rightChild == NULL) ? 0:node->rightChild->depth;

	return rightDepth - leftDepth;

}
static void LeftRotation(sd_TTreeNode**  rootNodePointer ,sd_TTreeNode*  pivotNode,  bool pivotNodeLock )
{   
	/*
	//X-Lock  the nodes related with the rotation
	//rootNode have been locked 
    //tempNode need be locked
	//pivotNode need be locked in the case of  RL. 
	In the case of  RR, there is once  rotation,the pivotNode has been locked in function    rebalance 
	*/
    sd_TTreeNode*  rootNode=*rootNodePointer;
	sd_TTreeNode*  tempNode;
	if(pivotNodeLock)
		LockExclusiveNode(pivotNode);
	LockExclusiveNode(pivotNode->leftChild);

	if(pivotNode->leftChild!=NULL)
		sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),pivotNode->leftChild->id);

	if (rootNode->parent == NULL)
		rootNode->tree->rootNode=pivotNode;
	else if(rootNode->parent->leftChild == rootNode)
		rootNode->parent->leftChild=pivotNode;
	else
		rootNode->parent->rightChild=pivotNode;
	
	tempNode= pivotNode->leftChild;

	if (rootNode->parent != NULL)
		pivotNode->parent = rootNode->parent;
	else
		pivotNode->parent = NULL;

	pivotNode->leftChild = rootNode;
	rootNode->parent = pivotNode;

	rootNode->rightChild = tempNode;
	
	if (tempNode != NULL)
		tempNode->parent = rootNode;
	
	rootNode= pivotNode; 
	*rootNodePointer=rootNode;
    
	if (tempNode != NULL)
		sd_log("%d unlock node %lld \n",sd_getCurrentThreadID(),tempNode->id);
	UnlockNode(tempNode);

	sd_log("%d unlock node %lld \n",sd_getCurrentThreadID(),rootNode->leftChild->id);
	UnlockNode(pivotNode->leftChild);

	UpdateNodeDepth(rootNode->leftChild);
	UpdateNodeDepth(rootNode->rightChild);
	UpdateNodeDepth(rootNode);
	/* 

	if ((*rootNodePointer)->parent == NULL)
		(*rootNodePointer)->tree->rootNode=pivotNode;
	else if((*rootNodePointer)->parent->leftChild == rootNode)
		(*rootNodePointer)->parent->leftChild=pivotNode;
	else
		(*rootNodePointer)->parent->rightChild=pivotNode;
	
	tempNode= pivotNode->leftChild;

	if ((*rootNodePointer)->parent != NULL)
		pivotNode->parent = (*rootNodePointer)->parent;
	else
		pivotNode->parent = NULL;

	pivotNode->leftChild = (*rootNodePointer);
	(*rootNodePointer)->parent = pivotNode;

	(*rootNodePointer)->rightChild = tempNode;
	
	if (tempNode != NULL)
		tempNode->parent = (*rootNodePointer);
	
	(*rootNodePointer)= pivotNode; 
	
	UnlockNode(tempNode);


	UnlockNode(pivotNode->leftChild);

	UpdateNodeDepth((*rootNodePointer)->leftChild);
	UpdateNodeDepth((*rootNodePointer)->rightChild);
	*/ 
}

static void RightRotation(sd_TTreeNode**  rootNodePointer ,sd_TTreeNode*  pivotNode, bool pivotNodeLock)
{
	/*
	//X-Lock  the nodes related with the rotation
	//rootNode have been locked 
    //tempNode need be locked
	//pivotNode need be locked in the case of  LR. 
	In the case of  LL, there is once  rotation,the pivotNode has been locked in function    rebalance 
	*/
	sd_TTreeNode*  rootNode=*rootNodePointer;
	sd_TTreeNode*  tempNode;
	if(pivotNodeLock)
		LockExclusiveNode(pivotNode);
	/*bugFix 2009-02-23 by wei
	when txnState isn't null && pivotNode->rightChild is null, there is a break*/
	LockExclusiveNode(pivotNode->rightChild);

	if(pivotNode->rightChild!=NULL)
		sd_log("%d lock node %lld \n",sd_getCurrentThreadID(),pivotNode->rightChild->id);


	if (rootNode->parent == NULL)
		rootNode->tree->rootNode=pivotNode;
	else if(rootNode->parent->leftChild == rootNode)
		rootNode->parent->leftChild=pivotNode;
	else
		rootNode->parent->rightChild=pivotNode;
	
	tempNode = pivotNode->rightChild;
	
	
	
	tempNode = pivotNode->rightChild;

	if (rootNode->parent != NULL)
		pivotNode->parent = rootNode->parent;
	else
		pivotNode->parent = NULL;

	pivotNode->rightChild = rootNode;
	rootNode->parent = pivotNode;

	rootNode->leftChild = tempNode;

	if (tempNode != NULL)
		tempNode->parent = rootNode;
	
	rootNode= pivotNode;

	*rootNodePointer=rootNode;
	if (tempNode != NULL)
		sd_log("%d unlock node %lld \n",sd_getCurrentThreadID(),tempNode->id);
	UnlockNode(tempNode);
	sd_log("%d unlock node %lld \n",sd_getCurrentThreadID(),rootNode->rightChild->id);
	UnlockNode(pivotNode->rightChild);

	UpdateNodeDepth(rootNode->leftChild);
	UpdateNodeDepth(rootNode->rightChild);
	UpdateNodeDepth(rootNode);
/*

	if ((*rootNodePointer)->parent == NULL)
		(*rootNodePointer)->tree->rootNode=pivotNode;
	else if((*rootNodePointer)->parent->leftChild == rootNode)
		(*rootNodePointer)->parent->leftChild=pivotNode;
	else
		(*rootNodePointer)->parent->rightChild=pivotNode;

	tempNode = pivotNode->rightChild;

	if (rootNode->parent != NULL)
		pivotNode->parent = (*rootNodePointer)->parent;
	else
		pivotNode->parent = NULL;

	pivotNode->rightChild = (*rootNodePointer);
	(*rootNodePointer)->parent = pivotNode;

	(*rootNodePointer)->leftChild = tempNode;

	if (tempNode != NULL)
		tempNode->parent = (*rootNodePointer);
	
	(*rootNodePointer)= pivotNode;



	UnlockNode(tempNode);

	UnlockNode(pivotNode->rightChild);

	UpdateNodeDepth((*rootNodePointer)->leftChild);
	UpdateNodeDepth((*rootNodePointer)->rightChild);*/
}

static void PrintNode(sd_TTreeNode* node, FILE* fp, int indent)
{
	int i;

	for (i=0;i<indent;i++)
		fprintf(fp,"\t");

	if (node == NULL)
	{
		fprintf(fp,"Node: NULL\n");
		return;
	}

	fprintf(fp,"Node (%lld): depth=%d, count=%d : ",node->id,node->depth, node->numRecords);
	for (i=0;i<(int)node->numRecords;i++)
	{
		sd_KeyPrint(&node->records[i].key, fp);
		if (i<(int)node->numRecords-1)
			fprintf(fp,", ");
	}
	fprintf(fp,"\n");
	if (node->leftChild != NULL || node->rightChild != NULL)
	{
		PrintNode(node->leftChild, fp, indent+1);
		PrintNode(node->rightChild, fp, indent+1);
	}
}

static sd_Ret CheckNode(sd_TTreeNode* node)
{
	int depth;
	int leftdepth, rightdepth;

	if (node == NULL)
		return SD_RET_OK;

	CheckNode(node->leftChild);
	CheckNode(node->rightChild);

	/* Check the depth */
	if (node->leftChild == NULL)
		leftdepth = 0;
	else
		leftdepth = node->leftChild->depth;
	if (node->rightChild == NULL)
		rightdepth = 0;
	else
		rightdepth = node->rightChild->depth;
	depth = MAX(leftdepth, rightdepth)+1;
	SD_ASSERT(depth == node->depth);

	/* Check the parent and child link pointer */
	if (node->leftChild != NULL)
	{
		SD_ASSERT(node == node->leftChild->parent);
	}
	if (node->rightChild != NULL)
	{
		SD_ASSERT(node == node->rightChild->parent);
	}

	return SD_RET_OK;
}

static sd_Ret LockShareNode(sd_TTreeNode* node)
{
	sd_TTree* tree = node->tree;
	if(node==NULL)
		return SD_RET_OK;
#ifdef _DEBUG_TTREE_LOCK
	{
		sd_uid nodeid;
		uint32 locktimes;
		uint32 tid;
		sd_Hashtable* nodetab;

		sd_SpinLock_Lock(&tree->tidLock);
		nodeid = node->id;
		tid = sd_getCurrentThreadID();
		if (sd_HashtableLookup(tree->nodeLockTable,&tid, &nodetab) == SD_RET_OK)
		{
			if (sd_HashtableLookup(nodetab, &nodeid, &locktimes) == SD_RET_OK)
			{
				locktimes++;
				sd_HashtableUpdate(nodetab,&nodeid, &locktimes);
			}
			else
			{
				locktimes=1;
				sd_HashtableInsert(nodetab, &nodeid, &locktimes);
			}
		}
		else
		{
			sd_HashtableCreate(&nodetab, 
				sd_HashFunc_uint64,
				sd_HashKeyCompare_uint64,
				7,
				8,
				sizeof(sd_uid),
				sizeof(uint32),
				NULL);
			locktimes = 1;
			sd_HashtableInsert(nodetab, &nodeid, &locktimes);
			sd_HashtableInsert(tree->nodeLockTable, &tid, &nodetab);
		}
		sd_SpinLock_Unlock(&tree->tidLock);
	}
	/*sd_log("TID %lld Rlock Node %d\n", tree->treeId,node->id);*/
#endif
	sd_RWLock_RLock(&node->lock);
	return SD_RET_OK;
}

static sd_Ret LockExclusiveNode(sd_TTreeNode* node)
{
	sd_TTree* tree = NULL;
	if(node==NULL)
		return SD_RET_OK;
	tree = node->tree;
#ifdef _DEBUG_TTREE_LOCK
	{
		sd_uid nodeid;
		uint32 locktimes;
		uint32 tid;
		sd_Hashtable* nodetab;

		sd_SpinLock_Lock(&tree->tidLock);
		nodeid = node->id;
		tid = sd_getCurrentThreadID();
		if (sd_HashtableLookup(tree->nodeLockTable,&tid, &nodetab) == SD_RET_OK)
		{
			if (sd_HashtableLookup(nodetab, &nodeid, &locktimes) == SD_RET_OK)
			{
				locktimes++;
				sd_HashtableUpdate(nodetab,&nodeid, &locktimes);
			}
			else
			{
				locktimes=1;
				sd_HashtableInsert(nodetab, &nodeid, &locktimes);
			}
		}
		else
		{
			sd_HashtableCreate(&nodetab, 
				sd_HashFunc_uint64,
				sd_HashKeyCompare_uint64,
				7,
				8,
				sizeof(sd_uid),
				sizeof(uint32),
				NULL);
			locktimes = 1;
			sd_HashtableInsert(nodetab, &nodeid, &locktimes);
			sd_HashtableInsert(tree->nodeLockTable, &tid, &nodetab);
		}
		sd_SpinLock_Unlock(&tree->tidLock);
	}
	/*sd_log("TID %lld Wlock Node %d\n", tree->treeId,node->id);*/
#endif
	sd_RWLock_WLock(&node->lock);
	return SD_RET_OK;
}

static sd_Ret UnlockNode(sd_TTreeNode* node)
{
	sd_TTree* tree;
	
	if(node==NULL)
		return SD_RET_OK;
	tree = node->tree;
#ifdef _DEBUG_TTREE_LOCK
	{
		sd_uid nodeid;
		uint32 locktimes;
		uint32 tid;
		sd_Hashtable* nodetab;
		sd_SpinLock_Lock(&tree->tidLock);
		nodeid = node->id;
		tid = sd_getCurrentThreadID();
		if (sd_HashtableLookup(tree->nodeLockTable,&tid, &nodetab) == SD_RET_OK)
		{
			if (sd_HashtableLookup(nodetab, &nodeid, &locktimes) == SD_RET_OK)
			{
				locktimes--;
				if (locktimes == 0)
				{
					sd_HashtableDelete(nodetab,&nodeid);
					if (nodetab->total_count == 0)
					{
						sd_HashtableFree(nodetab);
						sd_HashtableDelete(tree->nodeLockTable, &tid);
					}
				}
				else
					sd_HashtableUpdate(nodetab,&nodeid, &locktimes);
			}
			else
			{
				sd_HashtableIterator iter;
				if (nodetab->total_count > 0)
				{
					sd_HashtableBegin(nodetab,&iter);
					while(sd_HashtableNext(&iter) != SD_RET_FALSE)
					{
						uint32 nodeid;
						sd_HashtableGetKey(&iter, &nodeid);
						sd_log("TID %lld, %d, Unreleased node : %ld\n",tree->treeId,tid, nodeid);
					}
				}
				SD_ASSERT(false);
			}
		}
		else
		{
			sd_log("%d  unlock err  %lld \n",sd_getCurrentThreadID(),node->id);
			SD_ASSERT(false);
		}
		sd_SpinLock_Unlock(&tree->tidLock);
	}
	/*sd_log("TID %lld Unlock Node %d\n", tree->treeId,node->id);*/
#endif
	sd_RWLock_Unlock(&node->lock);
	return SD_RET_OK;
}

static void CheckLockTable(sd_TTree* tree)
{
	uint32 tid;
	sd_Hashtable* nodetab;
	sd_SpinLock_Lock(&tree->tidLock);
	tid = sd_getCurrentThreadID();
	if (sd_HashtableLookup(tree->nodeLockTable,&tid, &nodetab) == SD_RET_OK)
	{
		sd_HashtableIterator iter;
		if (nodetab->total_count > 0)
		{
			sd_HashtableBegin(nodetab,&iter);
			while(sd_HashtableNext(&iter) != SD_RET_FALSE)
			{
				sd_uid nodeid;
				sd_HashtableGetKey(&iter, &nodeid);
				sd_log("TID %lld, %d, Unreleased node : %lld\n",tree->treeId,tid, nodeid);
			}
			SD_ASSERT(false);
		}
	}
	sd_SpinLock_Unlock(&tree->tidLock);
}
